package com.gugee.tiktok.data.consumer.blogger.listener;

import com.alibaba.nacos.api.config.annotation.NacosValue;
import com.gugee.tiktok.data.common.msg.BloggerMessage;
import com.gugee.tiktok.data.common.msg.MarketMessage;
import com.gugee.tiktok.data.consumer.blogger.core.main.BloggerMarketDistributionMain;
import com.gugee.tiktok.data.consumer.blogger.core.main.UpdateBloggerMain;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 *
 * @author luolifeng
 * Date  2020/10/04
 */
@Component
public class MarketMessageListener {
    protected final Logger logger = LoggerFactory.getLogger(this.getClass());
    @Autowired
    BloggerMarketDistributionMain bloggerMarketDistributionMain;

    @NacosValue(value="${tiktok.data.blogger.accept.minTime.limit:1440}", autoRefreshed = true)
    long acceptMsgTimeMinRange;
//
//    @KafkaListener(id="${kafka.consumer.group.blogger.id}",topics = {"${kafka.topics.blogger.online}"},containerFactory = "factoryBloggerAck")
//    public void listen(List<MarketMessage> messages, Acknowledgment ack) {
//        long start=System.currentTimeMillis();
//        long acceptMsgTimeSecRange=acceptMsgTimeMinRange*60*1000;
//        String error="success";
//        try {
//            logger.info("接收到消息数量: {}" , messages.size());
//            long isRange=start-acceptMsgTimeSecRange;
//            /**
//             * 消息如果不在处理时限内，则丢弃
//             */
//            if(messages.get(0).getEventTime()>isRange) {
//                updateBloggerMain.handle(messages);
//            }else{
//                logger.info("消息数量: {} ,不在处理时限内，跳过;第一消息时间：{},时限分钟：{}" , messages.size(),messages.get(0).getEventTime(),acceptMsgTimeMinRange );
//            }
//            ack.acknowledge();
//        }catch (Exception e){
////            e.printStackTrace();
//            logger.info("发生异常: {}" , e.getMessage());
//            error=e.getMessage();
//        }finally {
//            long times=(System.currentTimeMillis()-start)/1000;
//            if(times<1){
//                times=1;
//            }
//            logger.info("消息数量: {},处理耗时: {}s,每秒处理消息数={}/s,结果：{}" , messages.size(),times,messages.size()/times,error);
//        }
//    }

    @KafkaListener(id="${kafka.consumer.group.market.id}",topics = {"${kafka.topics.market.online}"},containerFactory = "factoryMarketMsgAck")
    public void listen(ConsumerRecord<?, ?> record, Acknowledgment ack) {
        long start=System.currentTimeMillis();

            MarketMessage marketMessage=(MarketMessage)record.value();
//            String name = Thread.currentThread().getName();
//            System.out.println(name+"|"+String.format(
//                    "主题：%s，分区：%d，偏移量：%d，key：%s，value：%s",
//                    record.topic(),record.partition(),record.offset(),
//                    record.key(),record.value()));
        try {
            bloggerMarketDistributionMain.handle(marketMessage);

        }catch (Exception e){
            logger.error("MarketMessage 消息确认完毕！userId={},error={}",marketMessage.getAuthor().getCore_user_id(),e.getMessage());
        }finally {

            ack.acknowledge();
            logger.info("MarketMessage 消息确认完毕！userId={}",marketMessage.getAuthor().getCore_user_id());
        }
    }
}
